/*
Copyright (C) 2010- Peter Bui and Aaron Dingler.
This software is distributed under the GNU General Public License.
See the file COPYING for details.
*/

#include "smq_common.h"
#include "smq_message.h"
#include "smq_message_queue.h"
#include "smq_util.h"

#include <create_dir.h>
#include <debug.h>
#include <int_sizes.h>
#include <link.h>
#include <nvpair.h>
#include <sort_dir.h>
#include <stringtools.h>

#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <sys/types.h>

/* RPC functions ------------------------------------------------------------ */

int
smq_message_queue_create(const char *root, const char *name)
{
    char path[SMQ_LINE_MAX];

    snprintf(path, SMQ_LINE_MAX, "%s/%s/.bindings", root, name);

    if (!create_dir(path, 0700)) {
	debug(D_NOTICE, "cannot message queue %s: %s", name, strerror(errno));
	return 0;
    }

    return 1;
}

#define SMQ_MESSAGE_QUEUE_PUT_FILE(fname, fsize, failure) \
    fp = fopen((fname), "w"); \
    if (!fp) goto failure; \
    result = link_stream_to_file(client_link, fp, (fsize), time(0) + timeout); \
    if (result != fsize) goto failure; \
    fclose(fp); fp = NULL;

int
smq_message_queue_put(const char *root, const char *name, const INT64_T meta_size, const INT64_T body_size, struct link *client_link, int timeout)
{
    const char *uid;
    char temp_path[SMQ_LINE_MAX];
    char meta_path[SMQ_LINE_MAX];
    char body_path[SMQ_LINE_MAX];
    int result;
    FILE *fp = NULL;
    struct nvpair *nv = NULL;
    
    if (!smq_message_queue_exists(root, name)) return 0;

    uid = smq_util_get_unique_identifier();

    snprintf(temp_path, SMQ_LINE_MAX, "%s/%s/%s.temp", root, name, uid);
    snprintf(body_path, SMQ_LINE_MAX, "%s/%s/%s.body", root, name, uid);
    snprintf(meta_path, SMQ_LINE_MAX, "%s/%s/%s.meta", root, name, uid);

    debug(D_DEBUG, "getting meta %s", temp_path);
    SMQ_MESSAGE_QUEUE_PUT_FILE(temp_path, meta_size, smq_mqp_failure);

    debug(D_DEBUG, "parsing meta %s", temp_path);
    if ((nv = smq_message_read_header(temp_path)) == NULL) goto smq_mqp_failure;
    nvpair_delete(nv);

    debug(D_DEBUG, "getting body %s", body_path);
    SMQ_MESSAGE_QUEUE_PUT_FILE(body_path, body_size, smq_mqp_failure);
    
    debug(D_DEBUG, "renaming meta %s to %s", temp_path, meta_path);
    if (rename(temp_path, meta_path) < 0) goto smq_mqp_failure;

    link_write(client_link, uid, strlen(uid), time(0) + timeout);
    link_write(client_link, "\n", 1, time(0) + timeout);

    return 1;

smq_mqp_failure:
    if (nv) nvpair_delete(nv);
    if (fp) fclose(fp);
    unlink(temp_path);
    unlink(meta_path);
    unlink(body_path);
    return 0;
}

#define SMQ_MESSAGE_QUEUE_GET_FILE(type) \
    snprintf(file_path, SMQ_LINE_MAX,  "%s/%s/%s." type, root, name, mid); \
    stoptime  = time(0) + timeout; \
    file_size = smq_util_get_file_size(file_path); \
    fp = fopen(file_path, "r"); \
    debug(D_DEBUG, "getting %s." type, mid); \
    if (!fp) { \
	debug(D_NOTICE, "could not open %s", file_path); \
	result = 0; \
	snprintf(rpc_buffer, SMQ_LINE_MAX, "0 %s." type "\n", mid); \
	link_write(client_link, rpc_buffer, strlen(rpc_buffer), stoptime); \
    } else {\
	snprintf(rpc_buffer, SMQ_LINE_MAX, "%lld %s." type "\n", file_size, mid); \
	link_write(client_link, rpc_buffer, strlen(rpc_buffer), stoptime); \
	link_stream_from_file(client_link, fp, file_size, stoptime); \
	fclose(fp); fp = NULL; \
    }

int
smq_message_queue_get(const char *root, const char *name, const char *mid, struct link *client_link, int timeout)
{
    char file_path[SMQ_LINE_MAX];
    char rpc_buffer[SMQ_LINE_MAX];
    INT64_T file_size;
    int stoptime;
    FILE *fp = NULL;
    int result = 1;

    if (!smq_message_queue_exists(root, name)) return 0;

    SMQ_MESSAGE_QUEUE_GET_FILE("meta")
    SMQ_MESSAGE_QUEUE_GET_FILE("body")

    return result;
}

int
smq_message_queue_list(const char *root, const char *name, struct link *client_link, int timeout)
{
    int i;
    char queue_path[SMQ_LINE_MAX];
    char **messages;

    if (!smq_message_queue_exists(root, name)) return 0;

    snprintf(queue_path, SMQ_LINE_MAX, "%s/%s", root, name);
    if (!sort_dir(queue_path, &messages, strcmp)) return 0;

    for (i = 0; messages[i]; i++) {
	if (strstr(messages[i], ".meta")) {
	    link_write(client_link, messages[i], strlen(messages[i]) - 5, time(0) + timeout);
            link_write(client_link, " ", 1, time(0) + timeout);
	}
    }
    link_write(client_link, "\n", 1, time(0) + timeout);

    sort_dir_free(messages);

    return 1;
}

int
smq_message_queue_flush(const char *root, const char *name, struct link *client_link, int timeout)
{
    char flush_path[SMQ_LINE_MAX];
    FILE *fs;

    if (!smq_message_queue_exists(root, name)) return 0;

    snprintf(flush_path, SMQ_LINE_MAX, "%s/%s/.__flush", root, name);
    fs = fopen(flush_path, "w+");
    if (!fs) return 0;

    fclose(fs);
    return 1;
}

int
smq_message_queue_bind(const char *root, const char *name, const char *bind_name, const INT64_T bind_size, struct link *client_link, int timeout)
{
    char temp_path[SMQ_LINE_MAX];
    char bind_path[SMQ_LINE_MAX];
    int result;
    FILE *fp = NULL;
    
    if (!smq_message_queue_exists(root, name)) return 0;

    snprintf(temp_path, SMQ_LINE_MAX, "%s/%s/.bindings/%s.temp", root, name, bind_name);
    snprintf(bind_path, SMQ_LINE_MAX, "%s/%s/.bindings/%s", root, name, bind_name);

    if (smq_util_file_exists(bind_path)) return 0;
    
    SMQ_MESSAGE_QUEUE_PUT_FILE(temp_path, bind_size, smq_mqb_failure);
    if (rename(temp_path, bind_path) < 0) goto smq_mqb_failure;
    chmod(bind_path, 0700);

    return 1;

smq_mqb_failure:
    if (fp) fclose(fp);
    unlink(temp_path);
    unlink(bind_path);
    return 0;
}

int
smq_message_queue_unbind(const char *root, const char *name, const char *bind_name, struct link *client_link, int timeout)
{
    char bind_path[SMQ_LINE_MAX];
    
    if (!smq_message_queue_exists(root, name)) return 0;

    snprintf(bind_path, SMQ_LINE_MAX, "%s/%s/.bindings/%s", root, name, bind_name);

    if (!smq_util_file_exists(bind_path)) {
	goto smq_mqu_failure;
    } else {
	if (unlink(bind_path) < 0)
	    goto smq_mqu_failure;
    }
    
    return 1;

smq_mqu_failure:
    unlink(bind_path);
    return 0;
}

int
smq_message_queue_bindings(const char *root, const char *name, struct link *client_link, int timeout)
{
    int i;
    char bindings_path[SMQ_LINE_MAX];
    char **bindings;
    
    if (!smq_message_queue_exists(root, name)) return 0;

    snprintf(bindings_path, SMQ_LINE_MAX, "%s/%s/.bindings/", root, name);

    if (!smq_util_file_exists(bindings_path)) return 0;

    if (sort_dir(bindings_path, &bindings, strcmp)) {
	for (i = 0; bindings[i]; i++) {
	    if (bindings[i][0] != '.') {
		debug(D_DEBUG, "listing binding %s", bindings[i]);
		link_write(client_link, bindings[i], strlen(bindings[i]), time(0) + timeout);
                link_write(client_link, " ", 1, time(0) + timeout);
	    }
	}
	sort_dir_free(bindings);
    }
    link_write(client_link, "\n", 1, time(0) + timeout);
    
    return 1;
}

/* Public functions --------------------------------------------------------- */

int
smq_message_queue_count(const char *root, const char *name)
{
    int i;
    int count;
    char queue_path[SMQ_LINE_MAX];
    char **messages;

    if (!smq_message_queue_exists(root, name)) return -1;

    snprintf(queue_path, SMQ_LINE_MAX, "%s/%s", root, name);
    if (!sort_dir(queue_path, &messages, strcmp)) {
	debug(D_NOTICE, "smq_message_queue_count: sort_dir failed: %s", strerror(errno));
	return -1;
    }

    count = 0;
    for (i = 0; messages[i]; i++) 
	if (strstr(messages[i], ".meta")) 
	    count++;

    sort_dir_free(messages);

    return count;
}

int
smq_message_queue_exists(const char *root, const char *name)
{
    char path[SMQ_LINE_MAX];
    struct stat stats;
    
    snprintf(path, SMQ_LINE_MAX, "%s/%s", root, name);
    if (stat(path, &stats) < 0) return 0;
    
    return S_ISDIR(stats.st_mode);
}

/* 
vim: sw=4 sts=4 ts=8 ft=cpp
*/
